import decimal

import pandas as pd
from tqdm import tqdm

from components.config import WFQ_SOURCE_MYSQL_CONFIG
from utils.db.mysqldb import MysqlDB


class PddNewExport:
    def __init__(self):
        self._to_db = None

    @property
    def to_db(self):
        if self._to_db is None:
            self._to_db = MysqlDB(
                ip=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_IP"],
                port=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_PORT"],
                db=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_DB"],
                user_name=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_USER_NAME"],
                user_pass=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_USER_PASS"],
            )
        return self._to_db

    def get_data(self, params_list):
        params_str_list = [f" (mall_id = '{i['mall_id']}' and version = '{i['version']}') " for i in params_list]
        params_str = " or ".join(params_str_list)
        data_sql = f"""
        SELECT version as '月份',
        mall_id as '店铺ID',
        mall_name as '店铺名称',
        month_sales as '月销量（万件）',
        month_sales_money as '月销售额（万元）' ,
        total_sales as '累计销量（万件）',
        total_sales_money as '累计销售额（万元）',
        percent_month_sales as '月销量同比（%）',
        percent_month_sales_money as '月销售额同比（%）',
        percent_total_sales as '累计销量同比（%）',
        percent_total_sales_money as '累计销售额同比（%）',
        goods_num as '商品数（个）',
        mall_url as '店铺URL',
        mall_open_time_year as '店铺开店时长（年）',
        mall_status as '店铺状态',
        mall_desc as '店铺简介',
        mall_logo as '店铺图片链接',
        mall_star as '店铺综合体验评分',
        mall_label as '店铺认定资质',
        mall_details_label_list as '店铺服务标签',
        mall_achievements_rank as '榜单成就排名',
        mall_service_score as '服务评分'
         from zm_result_pdd_shop_info where {params_str}
        """

        detail_sql = f"""
        select version as '月份',
            mall_name as '店铺名称',
            mall_id as '店铺ID',
            goods_id as '商品ID',
            goods_name as '商品名称',
            price as '商品价格(元)',
            sale_num as '商品销量',
            goods_rank as '排名',
            hd_url as '商品图片url',
            market_price as '市场价格(元)',
            normal_price as '一般价格(元)',
            tag_list as '商品标签' from source_result_pdd_goods_detail where {params_str} """
        shop_data = self.to_db.find(data_sql, to_json=True)
        detail_data = self.to_db.find(detail_sql, to_json=True)
        shop_df = pd.DataFrame(shop_data)
        shop_df["月销量（万件）"] = shop_df["月销量（万件）"].apply(
            lambda x: (decimal.Decimal(x) / 10000) if x is not None else None
        )
        shop_df["月销售额（万元）"] = shop_df["月销售额（万元）"].apply(
            lambda x: (decimal.Decimal(x) / 10000) if x is not None else None
        )
        shop_df["累计销量（万件）"] = shop_df["累计销量（万件）"].apply(
            lambda x: (decimal.Decimal(x) / 10000) if x is not None else None
        )
        shop_df["累计销售额（万元）"] = shop_df["累计销售额（万元）"].apply(
            lambda x: (decimal.Decimal(x) / 10000) if x is not None else None
        )
        return shop_df, pd.DataFrame(detail_data)

    def run(self):
        sql = """
            select mall_id, version
            FROM zm_result_pdd_shop_info
            WHERE (mall_id, version) IN (
                    SELECT mall_id, MAX(version)
                FROM zm_result_pdd_shop_info where mall_name is not null and mall_name !=''
                GROUP BY mall_id
            )
        """
        params_list = self.to_db.find(sql, to_json=True)
        batch_size = 100
        shop_data = pd.DataFrame()
        detail_data = pd.DataFrame()

        import concurrent.futures

        with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
            futures = [
                executor.submit(
                    self.get_data,
                    params_list[i:i + batch_size],
                )
                for i in range(0, len(params_list), batch_size)
            ]
            for future in tqdm(concurrent.futures.as_completed(futures), total=len(futures)):
                if future.exception():
                    raise future.exception()
                shop_data_df, detail_data_df = future.result()
                shop_data = pd.concat([shop_data, shop_data_df])
                detail_data = pd.concat([detail_data, detail_data_df])

        # 每1w条保存一次
        save_size = 10000
        for i in range(8, len(shop_data), save_size):
            shop_data[i:i + 10000].to_csv(
                f"weifengqi_pdd_shop_202506_{str(i//save_size + 1).zfill(4)}.csv", index=False
            )

        for i in range(29, len(detail_data), 10000):
            detail_data[i:i + 10000].to_csv(
                f"weifengqi_pdd_product_202506_{str(i//save_size + 1).zfill(4)}.csv", index=False
            )


if __name__ == "__main__":
    PddNewExport().run()
